package cn.com.bluemoon.daps.model.service.impl;


import cn.com.bluemoon.daps.common.domain.ResultBean;
import cn.com.bluemoon.daps.common.domain.UserInfoHolder;
import cn.com.bluemoon.daps.common.enums.BmStatus;
import cn.com.bluemoon.daps.common.enums.DatabaseType;
import cn.com.bluemoon.daps.common.exception.DapException;
import cn.com.bluemoon.daps.common.toolkit.valueobject.MyQueryCallable;
import cn.com.bluemoon.daps.model.dto.DapDataModelTableDto;
import cn.com.bluemoon.daps.model.dto.ModelTableInfoDto;
import cn.com.bluemoon.daps.model.dto.SubordinateKeyDto;
import cn.com.bluemoon.daps.model.dto.TableListDto;
import cn.com.bluemoon.daps.model.dto.register.DapModelRegisterDto;
import cn.com.bluemoon.daps.model.entity.*;
import cn.com.bluemoon.daps.model.mapper.DapDataModelTableMapper;
import cn.com.bluemoon.daps.model.service.*;
import cn.com.bluemoon.daps.model.utils.factory.impl.table.AbsTableScript;
import cn.com.bluemoon.daps.model.utils.factory.impl.table.HiveTableScript;
import cn.com.bluemoon.daps.model.utils.factory.impl.table.MysqlTableScript;
import cn.com.bluemoon.daps.model.utils.factory.impl.table.PostgreTableScript;
import cn.com.bluemoon.daps.model.vo.DataModelTableInsertVO;
import cn.com.bluemoon.daps.model.vo.IdVo;
import cn.com.bluemoon.daps.model.vo.ModelIds;
import cn.com.bluemoon.daps.model.vo.group.in.GroupTableSearchVo;
import cn.com.bluemoon.daps.model.vo.register.SearchModelRegister;
import cn.com.bluemoon.daps.model.vo.share.SearchShareTableVo;
import cn.com.bluemoon.metadata.inter.RegisterDapDataModelTableService;
import cn.com.bluemoon.metadata.inter.dto.in.RegisterTableRequest;
import cn.com.bluemoon.metadata.sync.inter.SqlParseService;
import cn.com.bluemoon.metadata.sync.inter.dto.in.CheckSqlTypeRequest;
import cn.com.bluemoon.metadata.sync.inter.dto.out.security.CheckSqlTypeResultVo;
import com.alibaba.druid.DbType;
import com.alibaba.druid.sql.SQLUtils;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StopWatch;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.stream.Collectors;

import static org.apache.commons.lang3.StringUtils.isNotBlank;

/**
 * <p>
 * 数据模型-模型中设计表的表 服务实现类
 * </p>
 *
 * @author admin
 * @since 2020-08-03
 */
@Slf4j
@Service
public class DapDataModelTableServiceImpl extends ServiceImpl<DapDataModelTableMapper, DapDataModelTable> implements DapDataModelTableService {

    @Resource
    DapDataModelTableMapper dapDataModelTableMapper;

    @Resource
    DapDataModelTableFieldService dapDataModelTableFieldService;

    @Resource
    DapDataModelSubordinateKeyService dapDataModelSubordinateKeyService;

    @Resource
    DapDataModelTablePartitionService dapDataModelTablePartitionService;

    @Resource
    DapDataModelSystemTagMiddleService dapDataModelSystemTagMiddleService;

    @Resource
    DapDataModelRecordService dapDataModelRecordService;

    @Resource
    DapDataModelTableForeignKeyService dapDataModelTableForeignKeyService;

    @Resource
    DapDataModelService dapDataModelService;

    @Resource
    DapDataModelGroupService dapDataModelGroupService;

    @Resource
    RegisterDapDataModelTableService registerDapDataModelTableService;

    @Autowired
    private SqlParseService remoteSqlParseService;
    /**
     * 查询复制列表
     *
     * @param modelId
     * @return
     */
    @Override
    public List<DapDataModelTableDto> getCopyTableList(String modelId, Integer isNow) {
        List<DapDataModelTableDto> copyTableList = dapDataModelTableMapper.getCopyTableList(modelId, BmStatus.ON.getCode(), isNow);
        return copyTableList;
    }

    /**
     * 获取
     * @param modelId mId
     * @return tableId-> FieldId vs FieldName
     */
    @Override
    public Map<String, Map<String, String>> findNewestTablesPage(String modelId){
        return findNewestTablesPage(modelId, 0);
    }

    /**
     * 获取
     * @param modelId mId
     * @param pageBatchSize 1000
     * @return tableId-> FieldId vs FieldName
     */
    @Override
    public Map<String, Map<String, String>> findNewestTablesPage(String modelId, int pageBatchSize){
        // 获取模型的表总数n
        // 分一次pageSize条，然后合并
        int pageSize = 500;
        StopWatch sw = new StopWatch();
        sw.start("查询模型下的表Ids");
        LambdaQueryWrapper<DapDataModelTable> select = bmLambdaQuery().eq(DapDataModelTable::getDataModelId, modelId).select(DapDataModelTable::getId);
        List<String> tableIds = list(select).stream().map(DapDataModelTable::getId).collect(Collectors.toList());
        if (pageBatchSize <= 0) pageBatchSize = tableIds.size() < pageSize ? tableIds.size() / 2 : pageSize;
        sw.stop();
        sw.start("分片耗时");
        List<List<String>> partitions = Lists.partition(tableIds, pageBatchSize);
        sw.stop();
        sw.start("启动分片获取表字段信息");
        Map<String, Map<String, String>> tableIdAndFieldIdAndName = partitions.parallelStream()
                .filter(tbIds -> !tbIds.isEmpty())
                .flatMap(tbIds -> {
                    LambdaQueryWrapper<DapDataModelTableField> in = dapDataModelTableFieldService.bmLambdaQuery().eq(DapDataModelTableField::getDataModelId, modelId)
                            .eq(DapDataModelTableField::getPId, "0").eq(DapDataModelTableField::getDataNewest, 1)
                            .in(DapDataModelTableField::getModelTableId, tbIds);
                    List<DapDataModelTableField> fields = dapDataModelTableFieldService.list(in);
                    return fields.stream();
                }).collect(Collectors.groupingBy(DapDataModelTableField::getModelTableId,
                        Collectors.toMap(DapDataModelTableField::getId, DapDataModelTableField::getFieldName, (a, b) -> b)));
        sw.stop();
        log.info("获取模型{}的表信息情况：{}{}", modelId, System.lineSeparator(), sw.prettyPrint());
        return tableIdAndFieldIdAndName;
    }

    /**
     * 判断新增表名是否重复
     *
     * @param tableName
     * @return
     */
    @Override
    public DapDataModelTable getModelTableByName(String tableName, String modelId) {

        // 同模型下表名不允许重复
        DapDataModelTable table = dapDataModelTableMapper.selectOne(new LambdaQueryWrapper<DapDataModelTable>()
                .eq(DapDataModelTable::getEngName, tableName)
                .eq(DapDataModelTable::getBmStatus, BmStatus.ON.getCode())
                .eq(DapDataModelTable::getDataModelId, modelId)
                .eq(DapDataModelTable::getDataNewest, 1));
        return table;
    }

    /**
     * 新增或修改模型表
     *
     * @param vo
     * @return
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public ResultBean<String> insertModelTable(DataModelTableInsertVO vo) {

        // 首先判断英文表名在整个系统是否重复
        List<DapDataModelTable> modelTableByName = isRepeatTableName(vo.getDapDataModelTable().getEngName(), vo.getDapDataModelTable().getDataModelId());
        // id为空表示新增
        if (StringUtils.isBlank(vo.getDapDataModelTable().getId())) {
            // 为空进行插入操作
            if (modelTableByName.isEmpty()) {
                // 获得新增模型表id
                DapDataModelTable dapDataModelTable = vo.getDapDataModelTable();
                dapDataModelTable.setPId("0");
                // 是否为重建表
                String renewTable = getRenewTable(vo.getDapDataModelTable().getDataModelId(), vo.getDapDataModelTable().getEngName());
                if(renewTable != null){
                    dapDataModelTable.setId(renewTable);
                }
                dapDataModelTableMapper.insert(dapDataModelTable);
                // 判断是否为mysql导入hive的源表
                if(null != vo.getDapDataModelTable().getSourceTableId()){
                    // 更新为入仓状态
                    this.update(new LambdaUpdateWrapper<DapDataModelTable>()
                            .set(DapDataModelTable::getDataWarehouse, true)
                            .eq(DapDataModelTable::getId, vo.getDapDataModelTable().getSourceTableId()));
                }
                String id = dapDataModelTable.getId();
                // 首先判断标签列表是否为空，再进行引用标签
                if (vo.getSysTags().isEmpty()) {
                    return ResultBean.ok(id);
                }
                dapDataModelSystemTagMiddleService.insertModelTag(vo.getSysTags(), id, 1);
                return ResultBean.ok(id);
            } else {
                return ResultBean.error("系统中表英文名已存在，请重新输入");
            }
        } else {
            boolean flag = modelTableByName.stream().noneMatch(t -> t.getId().equals(vo.getDapDataModelTable().getId()));
            // 验证表名是否重复
            boolean isEdit = !modelTableByName.isEmpty() && flag ? false : true;
            if (isEdit) {
                this.updateById(vo.getDapDataModelTable());
                // 清除之前存在的标签
                dapDataModelSystemTagMiddleService.updateModelTag(vo.getDapDataModelTable().getId(), 1);
                if (!vo.getSysTags().isEmpty()) {
                    // 插入新增的标签
                    dapDataModelSystemTagMiddleService.insertModelTag(vo.getSysTags(), vo.getDapDataModelTable().getId(), 1);
                    return ResultBean.ok();
                }
                return ResultBean.ok();
            } else {
                return ResultBean.error("系统中表英文名已存在，请重新输入");
            }
        }
    }

    @Override
    public String getRenewTable(String modelId, String tableName){

        // 查询版本中存在是否存在已经恢复的表
        List<DapDataModelTable> tables = this.getBaseMapper().selectList(new LambdaQueryWrapper<DapDataModelTable>()
                .eq(DapDataModelTable::getEngName, tableName).eq(DapDataModelTable::getDataModelId, modelId)
                .ne(DapDataModelTable::getPId, "0").eq(DapDataModelTable::getBmStatus, BmStatus.ON.getCode()));
        if(!tables.isEmpty()){
            DapDataModelTable first = tables.stream().findFirst().get();
            return first.getPId();
        }
        return null;
    }

    /**
     * 注册列表
     * @param register
     * @return
     */
    @Override
    public IPage<DapModelRegisterDto> getVersionRegister(SearchModelRegister register) {

        Page<DapModelRegisterDto> page = new Page<>(register.getPageNum(), register.getPageSize());
        IPage<DapModelRegisterDto> pageList = dapDataModelTableMapper.selectModelVersionRegister(page, register);
        List<DapModelRegisterDto> tableList = Lists.newArrayList();
        if(!pageList.getRecords().isEmpty()){
            List<DapModelRegisterDto> records = pageList.getRecords();
            // 查询发布时间
            List<String> modelIds = records.stream().map(DapModelRegisterDto::getModelId).distinct().collect(Collectors.toList());
            for (String modelId : modelIds) {
                List<DapDataModelRecord> dapDataModelRecords = dapDataModelRecordService.getBaseMapper()
                        .selectList(new LambdaQueryWrapper<DapDataModelRecord>().eq(DapDataModelRecord::getModelId, modelId)
                                .eq(DapDataModelRecord::getPublicStatus, 1));
                LocalDateTime max = dapDataModelRecords.stream().filter(d -> d.getReleaseTime() != null)
                        .map(DapDataModelRecord::getReleaseTime).distinct().max((e1, e2) -> e1.compareTo(e2)).get();
                List<DapModelRegisterDto> collect = records.stream().filter(r -> r.getModelId().equals(modelId))
                        .peek(r -> r.setPublicTime(max)).collect(Collectors.toList());
                tableList.addAll(collect);
            }
            // 获取分组id
            List<String> idList = tableList.stream().map(DapModelRegisterDto::getGroupId).collect(Collectors.toList());
            // 查询分组信息
            List<DapDataModelGroup> dapDataModelGroups = dapDataModelGroupService.getBaseMapper().selectBatchIds(idList);
            // 筛选出未删除的分组
            List<DapDataModelGroup> childGroup = dapDataModelGroups.stream().filter(d -> d.getBmStatus().equals(BmStatus.ON)).collect(Collectors.toList());
            // 筛选出有父id的分组
            List<String> pidList = dapDataModelGroups.stream().filter(g -> isNotBlank(g.getPid()))
                    .map(DapDataModelGroup::getPid).collect(Collectors.toList());
            List<DapDataModelGroup> parentGroup = null;
            if(!pidList.isEmpty()){
                // 查询父分组信息
                List<DapDataModelGroup> list = dapDataModelGroupService.getBaseMapper().selectBatchIds(pidList);
                parentGroup = list.stream().filter(p -> p.getBmStatus().equals(BmStatus.ON)).collect(Collectors.toList());
            }
            // 父分组转化成map
            Map<String, String> parentName = parentGroup == null ? new HashMap<>() : parentGroup.stream()
                    .collect(Collectors.toMap(DapDataModelGroup::getId, DapDataModelGroup::getName));
            Map<String, String> groupAllPath = childGroup.stream().collect(Collectors.toMap(DapDataModelGroup::getId,
                    p -> String.format("%s/%s", StringUtils.isBlank(parentName.get(p.getPid())) ? "" : parentName.get(p.getPid()) , p.getName())));
            List<DapModelRegisterDto> collect = tableList.stream().peek(r -> r.setGroupPath(groupAllPath.get(r.getGroupId()))).collect(Collectors.toList());
            pageList.setRecords(collect);
        }
        return pageList;
    }

    @Override
    @Transactional
    public ResultBean<String> tableRegister(List<String> tableIds) {

        // 筛选已经注册的数据
        List<DapDataModelTable> dapDataModelTables = this.getBaseMapper().selectBatchIds(tableIds);
        List<String> names = dapDataModelTables.stream().filter(d -> d.getRegisterApi() == true)
                .map(DapDataModelTable::getEngName).collect(Collectors.toList());
        if(names.isEmpty()){
            List<DapDataModelTable> list = Lists.newArrayList();
            for(DapDataModelTable table : dapDataModelTables){
                // 版本数据
                DapDataModelTable versionTable = new DapDataModelTable();
                versionTable.setId(table.getId());
                versionTable.setRegisterApi(true);
                list.add(versionTable);
                // 最新数据
                DapDataModelTable newTable = new DapDataModelTable();
                newTable.setId(table.getPId());
                newTable.setRegisterApi(true);
                list.add(newTable);
            }
            // 获得到表pid
            List<String> collect = dapDataModelTables.stream().map(DapDataModelTable::getPId).collect(Collectors.toList());
            // 更新版本和最新数据
            Boolean flag = this.updateBatchById(list);
            if(flag){
                RegisterTableRequest registerTableRequest = new RegisterTableRequest();
                registerTableRequest.setPids(collect);
                cn.com.bluemoon.metadata.common.ResultBean resultBean = registerDapDataModelTableService.registerTable(registerTableRequest);
                if(resultBean.getCode() == 500){
                    List<String> ids = (List)resultBean.getContent();
                    List<String> tableNames = dapDataModelTables.stream().filter(d -> ids.contains(d.getPId()))
                            .map(DapDataModelTable::getEngName).collect(Collectors.toList());
                    return ResultBean.ok(String.format("表注册成功，元数据自动同步表失败，请联系管理员在[模型同步采集作业]添加表%s所在系统", tableNames.toString()));
                }
            }
            return flag ? ResultBean.ok("注册成功"): ResultBean.error("注册失败");
        }
        return ResultBean.error(String.format("注册失败，%s已注册", names));
    }

    @Override
    public ResultBean<String> tableRegisterCancel(List<String> tableIds) {

        // 筛选出未注册的数据
        List<DapDataModelTable> dapDataModelTables = this.getBaseMapper().selectBatchIds(tableIds);
        List<String> names = dapDataModelTables.stream().filter(d -> d.getRegisterApi() == false)
                .map(DapDataModelTable::getEngName).collect(Collectors.toList());
        if(names.isEmpty()){
            List<DapDataModelTable> list = Lists.newArrayList();
            for(DapDataModelTable table : dapDataModelTables){
                DapDataModelTable dapDataModelTable = new DapDataModelTable();
                dapDataModelTable.setId(table.getId());
                dapDataModelTable.setRegisterApi(false);
                list.add(dapDataModelTable);
                // 最新数据
                DapDataModelTable newTable = new DapDataModelTable();
                newTable.setId(table.getPId());
                newTable.setRegisterApi(false);
                list.add(newTable);
            }
            // 获得到表pid
            List<String> collect = dapDataModelTables.stream().map(DapDataModelTable::getPId).collect(Collectors.toList());
            Boolean flag = this.updateBatchById(list);
            if(flag){
                RegisterTableRequest registerTableRequest = new RegisterTableRequest();
                registerTableRequest.setPids(collect);
                cn.com.bluemoon.metadata.common.ResultBean resultBean = registerDapDataModelTableService.cancelRegisterTable(registerTableRequest);
                if(resultBean.getCode() == 500){
                    List<String> ids = (List)resultBean.getContent();
                    List<String> tableNames = dapDataModelTables.stream().filter(d -> ids.contains(d.getPId()))
                            .map(DapDataModelTable::getEngName).collect(Collectors.toList());
                    return ResultBean.ok(String.format("表取消注册成功，元数据自动同步表失败，请联系管理员在[模型同步采集作业]添加表%s所在系统", tableNames.toString()));
                }
            }
            return flag ? ResultBean.ok("取消注册成功") : ResultBean.error("取消注册失败");
        }
        return ResultBean.error(String.format("取消注册失败，%s未进行注册", names.toString()));
    }

    @Override
    public IPage<DapDataModelTable> getShareTableList(SearchShareTableVo vo, List<String> shareTableIds) {

        Page<DapDataModelTable> page = new Page<>(vo.getPageNum(), vo.getPageSize());
        if(shareTableIds.isEmpty() && Objects.equals(vo.getShareStatus(), 1)){
            return page;
        }
        LambdaQueryWrapper<DapDataModelTable> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(DapDataModelTable::getBmStatus, BmStatus.ON.getCode())
                .eq(DapDataModelTable::getDataModelId, vo.getModelId())
                .eq(DapDataModelTable::getPId, "0");
        if(StringUtils.isNotBlank(vo.getTableName())){
            wrapper.like(DapDataModelTable::getEngName, vo.getTableName());
        }
        if(StringUtils.isNotBlank(vo.getGroupId())){
            wrapper.eq(DapDataModelTable::getGroupId, vo.getGroupId());
        }
        if(!shareTableIds.isEmpty() && Objects.equals(vo.getShareStatus(), 0)){
            wrapper.notIn(DapDataModelTable::getId, shareTableIds);
        }
        if(!shareTableIds.isEmpty() && Objects.equals(vo.getShareStatus(), 1)){
            wrapper.in(DapDataModelTable::getId, shareTableIds);
        }
        return getBaseMapper().selectPage(page, wrapper);
    }

    @Override
    public List<String> getCreateTableSqls(ModelIds modelIds) {
        return doGetCreateTableSqls(modelIds);
    }

    @Override
    public void checkTableCreateSql(DatabaseType databaseType, String dataModelId, String tableId) throws DapException{
        doCheckTableCreateSql(databaseType, dataModelId, tableId);
    }

    private void doCheckTableCreateSql(DatabaseType databaseType, String dataModelId, String tableId) throws DapException {
        List<String> sqlKeywords = new ArrayList<>();
        try {
            ModelIds modelIds = new ModelIds();
            modelIds.setDatabaseType(databaseType);
            modelIds.setModelId(dataModelId);
            modelIds.setTableId(tableId);
            List<String> createTableSqls = getCreateTableSqls(modelIds);
            String createTableSql = createTableSqls.get(0);
            CheckSqlTypeRequest checkSqlTypeRequest = new CheckSqlTypeRequest();
            checkSqlTypeRequest.setSqlScript(createTableSql);
            //mysql hive pgsql impala
            checkSqlTypeRequest.setDatabaseType(getSqlParserDatabaseType(databaseType));
            cn.com.bluemoon.metadata.common.ResultBean<List<CheckSqlTypeResultVo>> sqlParserResult = remoteSqlParseService.checkSqlType(checkSqlTypeRequest);
            log.info("解析SQL情况：{}", sqlParserResult);
            if (sqlParserResult.isSuccess()) {
                List<CheckSqlTypeResultVo> checkSqlTypeResultVos = sqlParserResult.getContent();
                sqlKeywords = checkSqlTypeResultVos.stream().filter(v -> v.getKeywords() != null && !v.getKeywords().isEmpty())
                        .flatMap(v -> v.getKeywords().stream()).distinct().collect(Collectors.toList());
            }
        } catch (Exception e) {
            log.error("尝试对表语句进行关键字检查失败，忽略!", e);
        }
        if (!sqlKeywords.isEmpty()){
            // 异常则触发回滚
            throw new DapException(String.format("当前表名或字段名存在关键字【%s】", String.join(",", sqlKeywords)));
        }
    }

    private String getSqlParserDatabaseType(DatabaseType databaseType) {
        if (databaseType == null) return null;
        switch (databaseType){
            case MYSQL:
            case DORIS_DB:
                return "mysql";
            case POSTGRESQL:
            case PGSQL:
                return "pgsql";
            case HIVE:
                return "hive";
            case IMPALA:
                return "impala";
            default:
                return databaseType.getDbName();
        }
    }


    private List<String> doGetCreateTableSqls(ModelIds tables) {
        List<String> sqlList = new ArrayList<>();
        List<String> collect = Lists.newArrayList();
        if(org.apache.commons.lang.StringUtils.isNotBlank(tables.getTableId())){
            collect.add(tables.getTableId());
        }else {
            List<DapDataModelTable> list = getBaseMapper()
                    .selectList(new LambdaQueryWrapper<DapDataModelTable>()
                            .eq(DapDataModelTable::getDataModelId, tables.modelId)
                            .eq(DapDataModelTable::getBmStatus, BmStatus.ON.getCode())
                            .eq(DapDataModelTable::getPId, "0"));
            collect = list.stream().map(DapDataModelTable::getId).collect(Collectors.toList());
        }
        for (String modelTableId:collect) {
            // 根据模型表id查询数据库中的表和字段信息
            ModelTableInfoDto tableInfo = dapDataModelTableFieldService.getTableInfo(modelTableId);
            if(tableInfo == null){
                throw new DapException(String.format("表id%s不存在", modelTableId));
            }
            // 根据数据库类型获得sql
            if(tables.getDatabaseType().equals(DatabaseType.HIVE)){
                // 根据模型表id获取分区信息
                List<DapDataModelTablePartition> partitionList = dapDataModelTablePartitionService.getPartitionList(modelTableId);
                HiveTableScript hiveTableScript = new HiveTableScript();
                String hive = hiveTableScript.createTableScript(tableInfo, null, partitionList);
                String format = SQLUtils.format(hive, DbType.hive);
                sqlList.add(format);
            }else{
                // 根据模型表id获取索引信息
                List<SubordinateKeyDto> keyList = dapDataModelSubordinateKeyService.getKeyList(modelTableId);
                // 生成sql语句
                AbsTableScript absTableScript = tables.getDatabaseType().equals(DatabaseType.MYSQL) ? new MysqlTableScript() : new PostgreTableScript();
                String tableScript = absTableScript.createTableScript(tableInfo, keyList, null);
                DbType dbType = tables.getDatabaseType().equals(DatabaseType.MYSQL) ? DbType.mysql : DbType.postgresql;
                String format = SQLUtils.format(tableScript, dbType);
                sqlList.add(format);
            }
        }
        return sqlList;
    }

    /**
     * 判断表示是否在系统的模型中是否有重复
     * @param tableName
     * @param modelId
     * @return
     */
    @Override
    public List<DapDataModelTable> isRepeatTableName(String tableName, String modelId) {

        // 查询当前模型详情
        DapDataModel dapDataModel = dapDataModelService.getBaseMapper().selectById(modelId);
        // 查询该系统下的模型
        List<DapDataModel> dapDataModels = dapDataModelService.getModelListById(dapDataModel.getSystemId());
        // 模型id集合
        List<String> collect = dapDataModels.stream().map(DapDataModel::getId).collect(Collectors.toList());
        List<DapDataModelTable> table = dapDataModelTableMapper.selectList(new LambdaQueryWrapper<DapDataModelTable>()
                .eq(DapDataModelTable::getEngName, tableName)
                .eq(DapDataModelTable::getBmStatus, BmStatus.ON.getCode())
                .in(DapDataModelTable::getDataModelId, collect)
                .eq(DapDataModelTable::getDataNewest, 1));
        return table;
    }

    @Override
    public List<String> getRepeatTableName(List<String> tableNames, String modelId) {
        if (tableNames == null || tableNames.isEmpty()){
            return Collections.emptyList();
        }
        // 查询当前模型详情
        DapDataModel dapDataModel = dapDataModelService.getBaseMapper().selectById(modelId);
        // 查询该系统下的模型
        List<DapDataModel> dapDataModels = dapDataModelService.getModelListById(dapDataModel.getSystemId());
        // 模型id集合
        List<String> relateModelIds = dapDataModels.stream().map(DapDataModel::getId).collect(Collectors.toList());
        List<DapDataModelTable> table = dapDataModelTableMapper.selectList(new LambdaQueryWrapper<DapDataModelTable>()
                .in(DapDataModelTable::getEngName, tableNames)
                .eq(DapDataModelTable::getBmStatus, BmStatus.ON.getCode())
                .in(DapDataModelTable::getDataModelId, relateModelIds)
                .eq(DapDataModelTable::getDataNewest, 1));
        return table.stream().map(DapDataModelTable::getEngName).distinct().collect(Collectors.toList());
    }

    /**
     * 获得表详情
     *
     * @param modelId
     * @return
     */
    @Override
    public DapDataModelTable getModelTableById(String modelId) {

        DapDataModelTable dapDataModelTable = dapDataModelTableMapper.selectById(modelId);
        return dapDataModelTable;
    }

    /**
     * 更新表操作时间
     *
     * @param tableId
     */
    @Override
    public void saveTime(String tableId) {
        String account = UserInfoHolder.getAccount();
        dapDataModelTableMapper.saveUpdateTime(tableId, account);
    }

    /**
     * 删除备份表
     *
     * @param modelId
     */
    @Override
    public void deleteTable(String modelId) {
        dapDataModelTableMapper.delete(new LambdaQueryWrapper<DapDataModelTable>()
                .eq(DapDataModelTable::getDataModelId, modelId).ne(DapDataModelTable::getPId, "0"));
        // dapDataModelTableMapper.deleteTable(modelId);

    }

    /**
     * 删除模型中的表，可批量
     *
     * @param id
     * @return
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public Boolean deleteTableAll(String id, DatabaseType dataType) {

        // 查询表信息
        DapDataModelTable table = this.getBaseMapper().selectById(id);
        if(table == null){
            return false;
        }
        // 如果是目标表删除
        if(!table.getSourceTableId().equals("0")){
            this.update(new LambdaUpdateWrapper<DapDataModelTable>()
                    .set(DapDataModelTable::getDataWarehouse, false)
                    .eq(DapDataModelTable::getId, table.getSourceTableId())
                    .eq(DapDataModelTable::getBmStatus, BmStatus.ON.getCode()));
        }
        // 如果是源表删除
        if(table.getDataWarehouse()){
            this.update (new LambdaUpdateWrapper<DapDataModelTable>()
                    .set(DapDataModelTable::getSourceTableId, "0")
                    .eq(DapDataModelTable::getSourceTableId, table.getSourceTableId())
                    .eq(DapDataModelTable::getBmStatus, BmStatus.ON.getCode()));
        }
        List<String> ids = Arrays.asList(id);
        // 删除表引用的标签
        dapDataModelSystemTagMiddleService.deleteTableTags(ids);
        // 删除表
        dapDataModelTableMapper.deleteById(id);
        // 删除字段
        dapDataModelTableFieldService.getBaseMapper().delete(new LambdaQueryWrapper<DapDataModelTableField>()
                .eq(DapDataModelTableField::getModelTableId, id));
        // 如果为hive数据库
        if(dataType == DatabaseType.HIVE){
            // 删除分区
            dapDataModelTablePartitionService.getBaseMapper().delete(new LambdaQueryWrapper<DapDataModelTablePartition>()
                    .eq(DapDataModelTablePartition::getModelTableId, id));
        // 为pg和hive
        }else {
            // 删除索引
            dapDataModelSubordinateKeyService.getBaseMapper().delete(new LambdaQueryWrapper<DapDataModelSubordinateKey>()
                    .eq(DapDataModelSubordinateKey::getModelTableId, id));
        }
        // 删除外键关系
        dapDataModelTableForeignKeyService.delTableForeignKey(id);
        return true;
    }

    /**
     * 复制表
     *
     * @param vo
     * @return
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public ResultBean<?> copyTable(DataModelTableInsertVO vo) {

        // 查询表名是否重复
        List<DapDataModelTable> modelTableByName = isRepeatTableName(vo.getDapDataModelTable().getEngName(), vo.getDapDataModelTable().getDataModelId());
        // 为空表示可以进行新增
        if (modelTableByName.isEmpty()) {
            // 获得到该表下的字段列表
            List<DapDataModelTableField> fields = dapDataModelTableFieldService
                    .getBaseMapper().selectList(new LambdaQueryWrapper<DapDataModelTableField>()
                            .eq(DapDataModelTableField::getModelTableId, vo.getDapDataModelTable().getId())
                            .eq(DapDataModelTableField::getBmStatus, BmStatus.ON.getCode()));
            // 新增表操作
            DapDataModelTable dapDataModelTable = vo.getDapDataModelTable();
            dapDataModelTable.setId(null);
            dapDataModelTable.setDataWarehouse(false);
            this.getBaseMapper().insert(dapDataModelTable);
            // 新增标签
            Boolean aBoolean = dapDataModelSystemTagMiddleService.insertModelTag(vo.getSysTags(), dapDataModelTable.getId(), 1);
            // 新增字段，将字段的id置空
            if (!fields.isEmpty()) {
                fields.forEach(field -> {
                    field.setId(null);
                    field.setModelTableId(dapDataModelTable.getId());
                });
                boolean b = dapDataModelTableFieldService.saveBatch(fields);
                return ResultBean.ok(b);
            } else {
                return ResultBean.ok(aBoolean);
            }
        } else {
            return ResultBean.error("表名重复！");
        }
    }

    /**
     * 表下的所有信息 v1.2
     *
     * @param modelId
     * @return
     */
    @Override
    public List<DapDataModelTableDto> getCopyTableInfo(String modelId) {

        List<DapDataModelTableDto> copyTableInfo = dapDataModelTableMapper.getCopyTableInfo(modelId, BmStatus.ON.getCode());
        return copyTableInfo;
    }

    /**
     * v1.2版本获得分组下表数据
     *
     * @param ids
     * @return
     */
    @Override
    public List<TableListDto> getTableByGroupId(IdVo ids) {

        List<TableListDto> tableListDtos = dapDataModelTableMapper.selectTableByGroupId(ids, BmStatus.ON.getCode());
        return tableListDtos;
    }

    /**
     * 分组和表的关系
     * 数仓类：模型id+第一层groupid+第二层
     * 非数仓类：模型id+第一层groupId
     * <important>针对获取分组下的表信息中，需要增加查询条件获取模型最新的表数据</important>
     *
     * @param groupPid
     * @param groupId
     * @param modelId  模型id
     * @param
     * @return 表列表
     */
    @Override
    public List<DapDataModelTable> findNewestTablesByGroupPidAndGroupId(String groupPid, String groupId,
                                                                        String modelId) {
        List<DapDataModelTable> list = getBaseMapper().findNewestTablesByGroupPidAndGroupId(modelId, groupId);
        // 获取模型对应的最新发布时间，存在则比较表updateTime，不存在默认为change=true（无需循环一次list）
        Optional<LocalDateTime>  maxReleaseTime = dapDataModelRecordService.findMaxReleaseTimeByModelId(modelId);
        maxReleaseTime.ifPresent(releaseTime -> {
            if (!list.isEmpty()){
                list.parallelStream().forEach(t -> t.setChange(t.getUpdateTime().isAfter(releaseTime)));
            }
        });
        return list;
    }

    /**
     * 分组和表的关系
     * 数仓类：模型id+第一层groupid+第二层
     * 非数仓类：模型id+第一层groupId
     * 获取分组下的表数据量
     *
     * @param groupPid 组1层
     * @param groupId  组2层
     * @param modelId  模型id
     * @return 表列表
     */
    @Override
    public long countByGroupPidAndGroupIdAndDataNewest(String groupPid, String groupId, String modelId, boolean dataNewest) {
        return count(bmLambdaQuery()
                .eq(DapDataModelTable::getDataModelId, modelId)
                .eq(DapDataModelTable::getGroupPid, groupPid)
                .eq(DapDataModelTable::getGroupId, groupId)
                .eq(DapDataModelTable::getDataNewest, dataNewest ? 1 : 0)
        );
    }

    /**
     * 对表增加组关系
     *
     * @param tableIds 多表id
     * @param groupPid 组1
     * @param groupId  组2
     */
    @Override
    public void addGroupRelationToTables(List<String> tableIds, String groupPid, String groupId) {
        if (tableIds.isEmpty()) {
            log.warn("所选的表信息为空，不加入组");
            return;
        }
        LambdaUpdateWrapper<DapDataModelTable> set = bmLambdaUpdate()
                .in(DapDataModelTable::getId, tableIds)
                .set(DapDataModelTable::getGroupPid, groupPid)
                .set(DapDataModelTable::getGroupId, groupId);
        boolean update = update(set);
        log.info("更新表信息情况：{}，size:{}", update, tableIds.size());
    }

    /**
     * 分组获取当下的表数据
     * 分页
     * <important>针对获取分组下的表信息中，需要增加查询条件获取模型最新的表数据</important>
     *
     * @param currentGroup 当前组
     * @param searchVo     查询
     * @return
     */
    @SuppressWarnings({"unchecked", "rawtypes"})
    @Override
    public IPage<DapDataModelTable> pageDapDataModelTable(GroupTableSearchVo searchVo, DapDataModelGroup currentGroup) {
        LambdaQueryWrapper<DapDataModelTable> eq = bmLambdaQuery()
                .eq(DapDataModelTable::getDataNewest, 1)
                .eq(DapDataModelTable::getPId, "0")
                .eq(DapDataModelTable::getDataModelId, currentGroup.getModelId())
                .eq(isNotBlank(currentGroup.getPid()), DapDataModelTable::getGroupPid, currentGroup.getPid())
                .eq(DapDataModelTable::getGroupId, currentGroup.getId())
                .and(isNotBlank(searchVo.getKeyword()), (wrapper) -> wrapper
                        .like(isNotBlank(searchVo.getKeyword()), DapDataModelTable::getName, searchVo.getKeyword())
                        .or()
                        .like(isNotBlank(searchVo.getKeyword()), DapDataModelTable::getEngName, searchVo.getKeyword())
                );
        return page(searchVo.toMpPage(), eq);
    }

    @Override
    public List<TableListDto> getList(List<String> modelIds) {

        List<TableListDto> list = new ArrayList<>();
        for (String id : modelIds) {
            IdVo idVo = new IdVo();
            idVo.setModelId(id);
            List<TableListDto> tableByGroupId = getTableByGroupId(idVo);
            list.addAll(tableByGroupId);
        }
        return list;
    }

    /**
     * 获取模型下的最新？数据标
     *
     * @param modelId 模型
     * @param newest  最
     * @return
     */
    @Override
    public int countByModelIdAndNewest(String modelId, boolean newest) {
        final LambdaQueryWrapper<DapDataModelTable> eq = bmLambdaQuery().eq(DapDataModelTable::getDataModelId, modelId)
                .eq(DapDataModelTable::getDataNewest, (newest ? 1 : 0))
                .eq(newest, DapDataModelTable::getPId, "0");
        return count(eq);
    }

    /**
     * 获取模型下的最新？数据标
     *
     * @param modelId 模型
     * @param newest  最
     * @return
     */
    @Override
    public List<DapDataModelTable> findAllByPageByModelIdAndNewest(String modelId, boolean newest) {
        int limit = 1000;
        return multiDBQuery(new MyQueryCallable<DapDataModelTable>() {
            @Override
            public Callable<List<DapDataModelTable>> callTask(int offset, int limit) {
                return new Callable<List<DapDataModelTable>>() {
                    @Override
                    public List<DapDataModelTable> call() throws Exception {
                        return getBaseMapper(DapDataModelTableMapper.class).findByModelIdAndNewestLimit(modelId, newest, offset, limit);
                    }
                };
            }

            @Override
            public int count() {
                return countByModelIdAndNewest(modelId, newest);
            }
        }, limit);
    }
}
